package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.common.utils.CollectionUtils;
import com.alibaba.nacos.common.utils.StringUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.heima.common.redis.RedisCacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;

/**
 * @author ybw
 * @version 1.0
 * @date 2021/12/17 14:52
 */
@Service
@Transactional
@Slf4j
public class TaskServiceImpl implements TaskService {
  @Autowired private TaskinfoMapper taskinfoMapper;
  @Autowired private TaskinfoLogsMapper taskinfoLogsMapper;
  @Autowired private RedisCacheService redisCacheService;
  /**
   * 添加任务
   *
   * @param task
   * @return
   */
  @Override
  public Long addTask(Task task) {
    // 添加数据库任务表
    boolean flag = addTaskToDB(task);
    if (flag) {
      // 添加redis
      addTaskToRedis(task);
    }
    return task.getTaskId();
  }

  /**
   * 消费任务
   *
   * @param taskType
   * @param priority
   */
  @Override
  public Task pollTask(Integer taskType, Integer priority) {
    // 从redis的list中删除任务
    String listKey = "current_" + taskType + "_" + priority;
    String str = redisCacheService.lRightPop(listKey);

    if (StringUtils.isNotEmpty(str)) {
      // 修改数据库
      // 获取任务id
      Task task = JSON.parseObject(str, Task.class);
      taskinfoMapper.deleteById(task.getTaskId());
      // 修改日志状态
      TaskinfoLogs taskInfoLogs = new TaskinfoLogs();
      BeanUtils.copyProperties(task, taskInfoLogs);
      // 设置消费状态
      taskInfoLogs.setStatus(1);
      taskinfoLogsMapper.updateById(taskInfoLogs);
      return task;
    }
    return null;
  }

  /** 将zset刷新同步到list中 */
  @Override
  @Scheduled(cron = "0 0/1 * * * ? ")
  public void freshToList() {
    // 获取分布式锁
    String lock = redisCacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
    if (StringUtils.isNotEmpty(lock)) {
      log.info("{}拿到了分布式锁，用来将zset刷新同步到list中", Thread.currentThread().getName());
      // 获取zset中所有的key
      Set<String> zSetKeys = redisCacheService.scan("future_*");
      if (CollectionUtils.isNotEmpty(zSetKeys)) {
        for (String zSetKey : zSetKeys) {
          // 查询小于当前时间的value
          Set<String> scoreSet =
              redisCacheService.zRangeByScore(zSetKey, 0, System.currentTimeMillis());
          // 同步到list中
          if (CollectionUtils.isNotEmpty(scoreSet)) {
            String[] split = zSetKey.split("_");
            String listKey = "current_" + split[1] + "_" + split[2];
            redisCacheService.refreshWithPipeline(zSetKey, listKey, scoreSet);
          }
        }
      }
    }
  }

  /** 同步数据库数据到redis */
  @Override
  @Scheduled(cron = "0 0/5 * * * ? ")
  public void reloadData() {
    // 设置分布式锁
    String lock = redisCacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
    if (StringUtils.isNotEmpty(lock)) {
      log.info("{}拿到了分布式锁，用来将数据库数据同步到redis中", Thread.currentThread().getName());
      Set<String> currentKeys = redisCacheService.scan("current_*");
      Set<String> futureKeys = redisCacheService.scan("future_*");
      // 清理缓存
      redisCacheService.delete(currentKeys);
      redisCacheService.delete(futureKeys);
      // 查询数据库
      LambdaQueryWrapper<Taskinfo> qw = new LambdaQueryWrapper<>();
      // 获取未来5分钟
      Calendar calendar = Calendar.getInstance();
      calendar.add(Calendar.MINUTE, 5);
      qw.lt(Taskinfo::getExecuteTime, calendar.getTimeInMillis());
      List<Taskinfo> taskInfos = taskinfoMapper.selectList(qw);
      // 拷贝属性
      Task task = new Task();
      for (Taskinfo taskInfo : taskInfos) {
        BeanUtils.copyProperties(taskInfo, task);
        // 设置时间
        task.setExecuteTime(taskInfo.getExecuteTime().getTime());
        // 存入redis
        addTaskToRedis(task);
      }
    }
  }

  /**
   * 添加任务到redis中
   *
   * @param task
   */
  private void addTaskToRedis(Task task) {
    String listKey = "current_" + task.getTaskType() + "_" + task.getPriority();
    String zSetKey = "future_" + task.getTaskType() + "_" + task.getPriority();
    // 获取未来5分钟
    Calendar calendar = Calendar.getInstance();
    calendar.add(Calendar.MINUTE, 5);
    // 如果执行时间小于等于当前时间，存入list
    if (task.getExecuteTime() <= System.currentTimeMillis()) {
      redisCacheService.lLeftPush(listKey, JSON.toJSONString(task));
    } else if (task.getExecuteTime() <= calendar.getTimeInMillis()) {
      // 如果当前时间小于预设时间5分钟
      redisCacheService.zAdd(zSetKey, JSON.toJSONString(task), task.getExecuteTime());
    }
  }

  /**
   * 添加任务到数据库中
   *
   * @param task
   */
  private boolean addTaskToDB(Task task) {
    boolean flag = true;
    try {
      Taskinfo taskInfo = new Taskinfo();
      BeanUtils.copyProperties(task, taskInfo);
      // 日期格式转换
      taskInfo.setExecuteTime(new Date(task.getExecuteTime()));
      taskinfoMapper.insert(taskInfo);
      // 把任务id存入task中
      task.setTaskId(taskInfo.getTaskId());
      // 添加数据库任务日志表
      TaskinfoLogs taskInfoLogs = new TaskinfoLogs();
      BeanUtils.copyProperties(task, taskInfoLogs);
      // 初始化状态
      taskInfoLogs.setStatus(0);
      // 日期格式转换
      taskInfoLogs.setExecuteTime(new Date(task.getExecuteTime()));
      // 添加乐观锁version
      taskInfoLogs.setVersion(1);
      taskinfoLogsMapper.insert(taskInfoLogs);
    } catch (BeansException e) {
      e.printStackTrace();
      flag = false;
    }
    return flag;
  }
}
